package com.central.service

import java.sql.{DriverManager, Timestamp}
import java.util
import java.util.{Calendar, Date, Properties, UUID}

import com.alibaba.fastjson.{JSONArray, JSONObject}
import com.central.bean.WifiEntity
import com.central.controller.ActivityTime
import com.central.service
import com.central.utils.{ConfigUtil, DateFormat, LoadESData, ParseTime}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, SaveMode, SparkSession}
import org.elasticsearch.spark.rdd.EsSpark

import scala.util.{Failure, Success, Try}

class FollowService {
  val TABLE_FOLLOW_RESULT = ConfigUtil.getProperties().getProperty("table.follow.result")
  val TABLE_FOLLOW_DETAIL = ConfigUtil.getProperties().getProperty("table.follow.detail")
  val JDBC_URL = ConfigUtil.getProperties().getProperty("mysql.url")
  val FOLLOW_CODE_INDEX: String = ConfigUtil.getProperties().getProperty("follow.es.index")
  val FOLLOW_FACE_INDEX: String = ConfigUtil.getProperties().getProperty("follow.face.index")
  val FOLLOW_CAR_INDEX: String = ConfigUtil.getProperties().getProperty("follow.licenseplate.index")
  val TABLE_SITE_RELATION = ConfigUtil.getProperties().getProperty("table.site.relation")
  val TABLE_FOLLOW_STATUS = ConfigUtil.getProperties().getProperty("table.follow.status")
  val USER_NAME = ConfigUtil.getProperties().getProperty("mysql.username")
  val PASSWORD = ConfigUtil.getProperties().getProperty("mysql.password")



  def followAll(spark: SparkSession, jobId: String, conditions: FollowingSiteConditionEntity, threshold: Int) = {
    //    println(conditions.idNumber + "  " + conditions.slice + " " + conditions.beginTime + " " + conditions.endTime + " " + threshold + " " + jobId + "==follow param")
    import spark.implicits._
    val prop = new Properties()
    prop.put("user", "root")
    prop.put("password", "root")
    val frame: DataFrame = spark.read.jdbc(JDBC_URL, TABLE_SITE_RELATION, prop)
    val relate = frame.rdd.map(x => {
      //row 下标从0开始
      (x.get(2).toString, x.get(1).toString, x.get(3).toString)
    }).collect()
    val siteDevices: Map[String, Array[String]] = relate.groupBy(_._1).map(x => {
      val site = x._1
      val devices = x._2.map(_._2)
      (site, devices)
    })
    val hasliferow = conditions.hasliferow.toInt
    val fuzzyway = conditions.fuzzyway.toInt
    val activity = conditions.activityTimes
    val sites = conditions.siteNumbers.split(",")
    val devices = sites.flatMap(x => {
      siteDevices.get(x).get
    }).mkString(",")
    val transConditions: FollowingConditionEntity = FollowingConditionEntity(
      conditions.idType,
      conditions.idNumber,
      conditions.beginTime,
      conditions.endTime,
      conditions.slice,
      devices,
      conditions.hasliferow,
      conditions.fuzzyway,
      conditions.activityTimes
    )
    val esMessage: RDD[WifiEntityWithLongInsteadOfTimestamp] =
      if ("mac".equals(conditions.idType) || "imsi".equals(conditions.idType) || "imei".equals(conditions.idType)) {
        getFollowMessage(spark, FOLLOW_CODE_INDEX, transConditions)
      }
      else if ("face".equals(conditions.idType)) {
        getFollowMessage(spark, FOLLOW_FACE_INDEX, transConditions)
      }
      else {
        getFollowMessage(spark, FOLLOW_CAR_INDEX, transConditions)
      }

    if (esMessage != null) {
      //生活规律过滤
      val hasliferowRdd = if (hasliferow == 1) {
        val recentRdd =
          if ("mac".equals(conditions.idType) || "imsi".equals(conditions.idType) || "imei".equals(conditions.idType)) {
            LoadESData.loadRecentEs(spark, FOLLOW_CODE_INDEX, activity)
          }
          else if ("face".equals(conditions.idType)) {
            LoadESData.loadRecentEs(spark, FOLLOW_FACE_INDEX, activity)
          }
          else {
            LoadESData.loadRecentEs(spark, FOLLOW_CAR_INDEX, activity)
          }
        esMessage.map(x => {
          (x.deviceNumber, x)
        })
          .join(recentRdd)
          .map(x => {
            x._2._1
          })
      }
      else {
        esMessage
      }
      println(hasliferowRdd.count() + " ==========fellow has life count===========")
      val result: RDD[FollowResultWithDetail] = Try(hasliferowRdd.count()) match {

        case Success(count) if count > 0 => doFollowingAnalysisAsRDD(spark, jobId, transConditions, hasliferowRdd, threshold)
        case Failure(error) => { //如果加载hybase数据失败,则返回无碰撞结果的信息
          import spark.implicits._
          spark.sparkContext.parallelize(List(service.FollowResultWithDetail(jobId, "None", 0, error.getMessage, 0, "None", "None", "None", null)))
          //            .toDS()
        }
      }
      val count = result.count()
      //      println(count+"==========")
      if (count > 0) {
        //      log.infoLog(s"伴随分析结果", s"分析结果数量为:${result.count()}")
        //将结果保存到mysql当中
        val properties = new Properties()
        properties.put("user", "root")
        properties.put("password", "root")
        import spark.implicits._
        result
          .filter(x => {
            !x.codetype.equals("None")
          })

          .map(x => {
            FollowResult(x.resultid, x.codetype, x.count, x.createtime, x.jobid, x.tagcode, x.address,x.location, x.phone)
          })
          .toDF()
          .write.mode(SaveMode.Append).jdbc(JDBC_URL, TABLE_FOLLOW_RESULT, properties)
        result
          .filter(x => {
            !x.codetype.equals("None")
          })
          .flatMap(x => {
            x.detail.toArray
          })
          .map(x => {
            x.asInstanceOf[FollowDetail]
          })
          .filter(x => {
            !("None".equals(x.resultid)) || !("None".equals(x.codetype))
          })
          .toDF()
          //          .map(x => {
          //            x.detail
          //          })
          .write.mode(SaveMode.Append).jdbc(JDBC_URL, TABLE_FOLLOW_DETAIL, properties)
      }
    }
    updateStatus(jobId)
  }

  def updateStatus(jobId: String) = {
    val table = TABLE_FOLLOW_STATUS
    try {
      Class.forName("com.mysql.cj.jdbc.Driver")
    } catch {
      case e =>
        e.printStackTrace()
    }
    val connection = DriverManager.getConnection(JDBC_URL, USER_NAME, PASSWORD)
    val statement = connection.prepareStatement(
      s"update  $table set status = ?, successtime=? where id = ?  "
    )
    statement.setInt(1, 2)
    statement.setString(2, ParseTime.LongToDate(System.currentTimeMillis().toString))
    statement.setInt(3, jobId.toInt)
    statement.executeUpdate()
    statement.close()
    connection.close()
  }

  def getFollowMessage(spark: SparkSession, index: String, conditions: FollowingConditionEntity): RDD[WifiEntityWithLongInsteadOfTimestamp] = {

    require(conditions != null)
    val startTime = DateFormat.dateToString(new Date(conditions.beginTime))
    val endTime = DateFormat.dateToString(new Date(conditions.endTime))
    var deviceCodes: util.ArrayList[String] = new util.ArrayList()
    conditions.deviceNumbers.split(",").map(s => deviceCodes.add(s))
    //    conditions.deviceNumbers.split(",").toList
    var flowRDD: RDD[WifiEntity] = null;

    //根据伴随条件，从es中查询数据得到相应的rdd
    val esRdd: RDD[(String, collection.Map[String, AnyRef])] = LoadESData.getMutliDeviceData(spark, index, deviceCodes, startTime, endTime)

    //对查询到的结果进行封装
    val esRddCount = esRdd.count()
    if (esRddCount > 0) {
      if (flowRDD == null) {
        flowRDD = esRdd.map(e => {
          if (e._2.get("mac").getOrElse(None) != None) //mac数据的封装
            WifiEntity(
              e._1, e._2.get("mac").get + "",
              e._2.get("mac").get + "",
              "mac",
              Timestamp.valueOf(e._2.get("captime").get + ""),
              Timestamp.valueOf(e._2.get("captime").get + ""),
              //            Timestamp.valueOf(e._2.get("ir_time").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),

              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").getOrElse("") + "",
              e._2.get("location").getOrElse("") + "",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),
              e._2.get("phone").getOrElse("") + "",
              "",
              e._2.get("picpath").getOrElse("") + ""
            )
          else if (e._2.get("imsi").getOrElse(None) != None)
            WifiEntity(
              e._1, e._2.get("imsi").get + "", e._2.get("imsi").get + "", "imsi",
              Timestamp.valueOf(e._2.get("captime").get + ""), Timestamp.valueOf(e._2.get("captime").get + ""),
              //            Timestamp.valueOf(e._2.get("ir_time").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").getOrElse("") + "",
              e._2.get("location").getOrElse("") + "",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),

              e._2.get("phone").getOrElse("") + "",
              "",
              e._2.get("picpath").getOrElse("") + ""
            )
          else if (e._2.get("imei").getOrElse(None) != None)
            WifiEntity(
              e._1, e._2.get("imei").get + "", e._2.get("imei").get + "", "imei",
              Timestamp.valueOf(e._2.get("captime").get + ""), Timestamp.valueOf(e._2.get("captime").get + ""),
              //                Timestamp.valueOf(e._2.get("ir_time").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").getOrElse("") + "",
              e._2.get("location").getOrElse("") + "",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),

              e._2.get("phone").getOrElse("") + "", "",
              e._2.get("picpath").getOrElse("") + ""
            )
          else if (e._2.get("faceid").getOrElse(None) != None)
            WifiEntity(
              e._1,
              e._2.get("faceid").get + "",
              e._2.get("faceid").get + "",
              "face",
              Timestamp.valueOf(e._2.get("captime").get + ""),
              Timestamp.valueOf(e._2.get("captime").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").getOrElse("") + "",
              e._2.get("location").getOrElse("") + "",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),
              e._2.get("phone").getOrElse("") + "",
              "",
              e._2.get("facepath").getOrElse("") + ""
            )
          else
            WifiEntity(
              e._1,
              e._2.get("licenseplateid").get + "",
              e._2.get("licenseplateid").get + "",
              "licenseplate",
              Timestamp.valueOf(e._2.get("captime").get + ""),
              Timestamp.valueOf(e._2.get("captime").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").getOrElse("") + "",
              e._2.get("location").getOrElse("") + "",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),
              e._2.get("phone").getOrElse("") + "",
              "",
              e._2.get("licenseplate_path").getOrElse("") + ""
            )
        })

      }
      else {
        flowRDD = flowRDD ++ esRdd.map(e => {
          if (e._2.get("mac").getOrElse(None) != None) //mac数据的封装
            WifiEntity(
              e._1, e._2.get("mac").get + "",
              e._2.get("mac").get + "",
              "mac",
              Timestamp.valueOf(e._2.get("captime").get + ""),
              Timestamp.valueOf(e._2.get("captime").get + ""),
              //            Timestamp.valueOf(e._2.get("ir_time").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),

              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").getOrElse("") + "",
              e._2.get("location").getOrElse("") + "",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),


              e._2.get("phone").getOrElse("") + "",
              "",
              e._2.get("picpath").getOrElse("") + ""
            )
          else if (e._2.get("imsi").getOrElse(None) != None)
            WifiEntity(
              e._1, e._2.get("imsi").get + "", e._2.get("imsi").get + "", "imsi",
              Timestamp.valueOf(e._2.get("captime").get + ""), Timestamp.valueOf(e._2.get("captime").get + ""),
              //            Timestamp.valueOf(e._2.get("ir_time").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").getOrElse("") + "",
              e._2.get("location").getOrElse("") + "",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),

              e._2.get("phone").getOrElse("") + "",
              "",
              e._2.get("picpath").getOrElse("") + ""
            )
          else if (e._2.get("imei").getOrElse(None) != None)
            WifiEntity(
              e._1, e._2.get("imei").get + "", e._2.get("imei").get + "", "imei",
              Timestamp.valueOf(e._2.get("captime").get + ""), Timestamp.valueOf(e._2.get("captime").get + ""),
              //                Timestamp.valueOf(e._2.get("ir_time").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").getOrElse("") + "",
              e._2.get("location").getOrElse("") + "",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),

              e._2.get("phone").getOrElse("") + "", "",
              e._2.get("picpath").getOrElse("") + ""
            )
          else if (e._2.get("faceid").getOrElse(None) != None)
            WifiEntity(
              e._1,
              e._2.get("faceid").get + "",
              e._2.get("faceid").get + "",
              "face",
              Timestamp.valueOf(e._2.get("captime").get + ""),
              Timestamp.valueOf(e._2.get("captime").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").getOrElse("") + "",
              e._2.get("location").getOrElse("") + "",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),
              e._2.get("phone").getOrElse("") + "",
              "",
              e._2.get("facepath").getOrElse("") + ""
            )
          else
            WifiEntity(
              e._1,
              e._2.get("licenseplateid").get + "",
              e._2.get("licenseplateid").get + "",
              "licenseplate",
              Timestamp.valueOf(e._2.get("captime").get + ""),
              Timestamp.valueOf(e._2.get("captime").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").getOrElse("") + "",
              e._2.get("location").getOrElse("") + "",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),
              e._2.get("phone").getOrElse("") + "",
              "",
              e._2.get("licenseplate_path").get + ""
            )
        })
      }
    }
    else {
      println("未找到相应的数据")
    }

    //对结果进行封装
    println("=========================")
    if (flowRDD != null) {
      val rdd: RDD[WifiEntityWithLongInsteadOfTimestamp] = flowRDD
        .mapPartitions(x => {
          x.map(wifiEntity =>
            WifiEntityWithLongInsteadOfTimestamp(
              wifiEntity.id,
              wifiEntity.showedIdNumber,
              wifiEntity.idNumber,
              wifiEntity.idNumberType,
              wifiEntity.behaviorStartTime.getTime,
              wifiEntity.behaviorEndTime.getTime,
              wifiEntity.deviceNumber,
              wifiEntity.address,
              wifiEntity.location,
              wifiEntity.deviceName,
              wifiEntity.captime,
              wifiEntity.phoneNumber,
              wifiEntity.latitude,
              wifiEntity.longitude,
              wifiEntity.picpath
            ))
        })
      return rdd
    }
    else null
  }

  def doFollowingAnalysisAsRDD(spark: SparkSession,
                               jobId: String,
                               conditions: FollowingConditionEntity,
                               wifiEntities: RDD[WifiEntityWithLongInsteadOfTimestamp],
                               threshold: Int): RDD[FollowResultWithDetail] = {
    val fuzzyway = conditions.fuzzyway.toInt

    val followingConditions = wifiEntities
      .filter(entity => entity.idNumber.equals(conditions.idNumber))
      .mapPartitions(x => {
        x.map(entity =>
          (entity.deviceNumber, entity.behaviorStartTime - conditions.slice * 1000, entity.behaviorStartTime + conditions.slice * 1000)
        )
      })
      .collect()
    println(followingConditions.size + "followingConditionsCount")
    val finalResultRdd =
      if (followingConditions.isEmpty) {
        spark.sparkContext.parallelize(List(FollowResultWithDetail(jobId, "None", 0, "", 0, "None", "None", "None", null)))
      }
      else {
        val followingBehaviorConditions = spark.sparkContext.broadcast(followingConditions)
        val value: RDD[(String, WifiEntityWithLongInsteadOfTimestamp)] = wifiEntities
          .filter(
            entity => {
              //得到所有符合伴随条件的人,即在所有具体时间范围内在需要分析的点位出现的次数大于0的数据。
              val broadcastConditions = followingBehaviorConditions.value
              broadcastConditions
                .filter {
                  case (device, start, end) => {
                    (!entity.idNumber.equals(conditions.idNumber)) &&
                      entity.deviceNumber.equals(device) && entity.behaviorStartTime >= start && entity.behaviorStartTime <= end
                  }
                }
                .size > 0
            })
          .map(entity => (entity.idNumber, entity))

        val groupRdd = value
          .aggregateByKey(List.empty[WifiEntityWithLongInsteadOfTimestamp])(
            (items, item) => items :+ item,
            (listOne, listAnother) => {
              listOne.foldLeft(listAnother)((items, item) => {
                items :+ item
              })
            })
          .map(x => {
            (x._1, x._2, x._2.size)
          })
        //模糊处理
        println("================fellow 模糊=================")
        val fuzzyRdd = if (fuzzyway != 0 && ("mac".equals(conditions.idType) || "imsi".equals(conditions.idType) || "imei".equals(conditions.idType))) {
          val all_devices = conditions.deviceNumbers.split(",")
          //
          val fuzzywayResult = groupRdd.map(x => {
            val code = x._1
            val count = x._2.size
            val detail = x._2
            val device = detail.map(x => {
              x.deviceNumber
            })
            val code_type = x._2.head.idNumberType

            (code, count, code_type, device, detail)
          })
            .collect()
            .map(data => {
              val code = data._1
              var count = data._2
              val devices = data._4
              //侦码出现的点位集
              val code_type = data._3
              //只对数据集个数小于阈值、且相差不超过2的情况进行模糊处理
              if (count < threshold && (threshold - count < 2)) {
                //所有条件点位集与 单个侦码出现过的点位集的差集。<如果这个`差集`  与  `模糊查询得到的缺失点位集` 有交集，则表示点位缺失。>
                val diff_devices = all_devices.diff(devices)
                if (fuzzyway == 1) {
                  println("========fellow type1==========")
                  var relat_code = findRelatDevice(spark, FOLLOW_CODE_INDEX, code)
                  if (diff_devices.intersect(relat_code).size > 0) {
                    count = count + 1
                  }
                  (data._1, data._5, count)
                }
                else if (fuzzyway == 2) {
                  println("========fellow type1==========")
                  var relat_code = findTrailDevice(spark, FOLLOW_CODE_INDEX, code, code_type)
                  if (diff_devices.intersect(relat_code).size > 0) {
                    count = count + 1
                  }
                  (data._1, data._5, count)
                }
                else {
                  println("========fellow type1==========")
                  var relat_code = findHotDevice(spark, FOLLOW_CODE_INDEX, code, code_type)
                  if (diff_devices.intersect(relat_code).size > 0) {
                    count = count + 1
                  }
                  (data._1, data._5, count)
                }
              }
              else {
                (data._1, data._5, count)
              }
            })
          spark.sparkContext.parallelize(fuzzywayResult)
        }
        else {
          groupRdd
        }

        val resultRdd =
          fuzzyRdd

            .filter(_._3 >= threshold).cache()
            .map {

              case (idNumber, entities, count) => {
                val id = UUID.randomUUID() + "123"
                val head = entities.head
                //对所有关系人所出现的设备号进行排序,如果出现在的设备地址不足阀值,则舍弃
                val detailsGroupByDeviceNumber = entities.groupBy(entity => entity.deviceNumber)
                if (detailsGroupByDeviceNumber.keySet.size >= threshold) {

                  //                Some(FollowingResult(jobId, idNumber, entities.head.idNumberType, entities.size, entities.map(_.id).mkString(",")))

                  val followDetails = entities.map(x => {

                    FollowDetail(
                      x.captime,
                      x.idNumberType,
                      x.deviceNumber.toInt,
                      x.devicename,
                      x.address,
                      x.phone,
                      id,
                      x.idNumber,
                      x.latitude,
                      x.longitude,
                      x.picpath
                    )
                  })

                  val details = new util.ArrayList[FollowDetail]()
                  for (elem <- followDetails) {
                    details.add(elem)
                  }

                  Some(

                    FollowResultWithDetail(
                      id,
                      entities.head.idNumberType,
                      //                    entities.map(_.id).mkString(","),
                      entities.size,
                      ParseTime.LongToDate(System.currentTimeMillis() + ""),
                      jobId.toInt,
                      entities.head.idNumber,
                      entities.head.address,
                      entities.head.location,
                      entities.head.phone,
                      details
                    )
                  )
                }
                else
                  None
              }
            }
            .cache()
            .filter(result => result.isDefined)
            .map(_.get)
        //        log.infoLog("!!!!!!!!!!!!!!!!!!!!", s"存在${resultRdd.count()}个人存在伴随关系")
        if (resultRdd.isEmpty())
          spark.sparkContext.parallelize(List(FollowResultWithDetail(jobId, "None", 0, ParseTime.LongToDate(System.currentTimeMillis() + ""), 0, "None", "None", "None", null)))
        else
          resultRdd
      }
    //    import spark.implicits._
    //    finalResultRdd.toDS()
    finalResultRdd
  }


  def findRelatDevice(spark: SparkSession, index: String, code: String) = {

    //找出code关联的号码
    try {
      //      Class.forName("com.mysql.jdbc.Driver")
      Class.forName("com.mysql.cj.jdbc.Driver")
    } catch {
      case e =>
        e
    }
    var relatcode = new util.ArrayList[String]()
    val connection = DriverManager.getConnection(JDBC_URL, USER_NAME, PASSWORD)
    val statement = connection.prepareStatement(s"select tagcode,tagtype from t_wifi_eachother_relation where srccode=$code order by counts desc")
    val resultSet = statement.executeQuery()
    var data = ""
    if (resultSet.next()) {
      data = (resultSet.getString(1) + "_" + resultSet.getString(2))
    }
    resultSet.close()
    statement.close()
    connection.close()

    if (data.length > 0) {
      val codetype = data.split("_")(0)
      val tagcodes = data.split("_")(1)

      val queryPara = new JSONObject()
      val queryBool = new JSONObject()
      val queryCode = new JSONObject()
      if (codetype.equals("imsi")) {
        queryCode.put("imsi", tagcodes)
      }
      else if (codetype.equals("mac")) {
        queryCode.put("mac", tagcodes)
      }
      else {
        queryCode.put("imei", tagcodes)
      }

      val calendar = Calendar.getInstance()
      val date = new Date()
      calendar.setTime(date)
      calendar.add(Calendar.DATE, 0 - 10)
      calendar.set(Calendar.HOUR, 0)
      calendar.set(Calendar.AM_PM, 0)
      calendar.set(Calendar.MINUTE, 0)
      calendar.set(Calendar.SECOND, 0)
      val starttime = ParseTime.sdf.format(calendar.getTime)
      calendar.add(Calendar.DATE, 10)
      val endtime = ParseTime.sdf.format(calendar.getTime)


      val queryTerms = new JSONObject()
      queryTerms.put("term", queryCode)
      val queryFilters = new JSONArray()
      queryFilters.add(queryTerms)
      val queryTime = new JSONObject()
      queryTime.put("gte", starttime)
      queryTime.put("lte", endtime)
      val querycaptime = new JSONObject()
      querycaptime.put("captime", queryTime)
      val queryRange = new JSONObject()
      queryRange.put("range", querycaptime)
      queryFilters.add(queryRange)

      queryBool.put("filter", queryFilters)
      queryPara.put("bool", queryBool)
      val esRdd = EsSpark.esRDD(spark.sparkContext, index, queryPara.toJSONString)
      esRdd.map(x => {
        x._2.get("device_code").get + ""
      })
        .map(x => {
          (x, 1)
        })
        .reduceByKey(_ + _)
        .collect()
        .sortBy(x => {
          x._2
        })
        .reverse
        .map(x => {
          x._1
        })
        .splitAt(3)
        ._1
    }
    else null
  }


  def findHotDevice(spark: SparkSession, index: String, code: String, codetype: String) = {

    val calendar = Calendar.getInstance()
    val date = new Date()
    calendar.setTime(date)
    calendar.add(Calendar.DATE, 0 - 10)
    calendar.set(Calendar.HOUR, 0)
    calendar.set(Calendar.AM_PM, 0)
    calendar.set(Calendar.MINUTE, 0)
    calendar.set(Calendar.SECOND, 0)
    val starttime = ParseTime.sdf.format(calendar.getTime)
    calendar.add(Calendar.DATE, 10)
    val endtime = ParseTime.sdf.format(calendar.getTime)

    val queryPara = new JSONObject()
    val queryBool = new JSONObject()

    val queryMatch = new JSONObject()
    queryMatch.put(codetype, code)
    val queryMast = new JSONObject()
    queryMast.put("match", queryMatch)
    queryBool.put("must", queryMast)

    val queryTime = new JSONObject()
    queryTime.put("gte", starttime)
    queryTime.put("lte", endtime)
    val querycaptime = new JSONObject()
    querycaptime.put("captime", queryTime)
    val queryRange = new JSONObject()
    queryRange.put("range", querycaptime)
    val queryMasts = new JSONArray()
    queryMasts.add(queryRange)
    val queryBools = new JSONObject()
    queryBools.put("must", queryMasts)
    val queryFilter = new JSONObject()
    queryFilter.put("bool", queryBools)
    queryBool.put("filter", queryFilter)
    queryPara.put("bool", queryBool)
    val esRdd = EsSpark.esRDD(spark.sparkContext, index, queryPara.toJSONString)

    esRdd.map(x => {
      (x._2.get("device_code").get + "", 1)
    })
      .reduceByKey(_ + _)
      .collect()
      .sortBy(x => {
        x._2
      })
      .reverse
      .splitAt(3)
      ._1
      .map(x => {
        x._1
      })
  }

  def findTrailDevice(spark: SparkSession, index: String, code: String, codetype: String) = {
    val calendar = Calendar.getInstance()
    val date = new Date()
    calendar.setTime(date)
    calendar.add(Calendar.DATE, 0 - 10)
    calendar.set(Calendar.HOUR, 0)
    calendar.set(Calendar.AM_PM, 0)
    calendar.set(Calendar.MINUTE, 0)
    calendar.set(Calendar.SECOND, 0)
    val starttime = ParseTime.sdf.format(calendar.getTime)
    calendar.add(Calendar.DATE, 10)
    val endtime = ParseTime.sdf.format(calendar.getTime)

    val queryPara = new JSONObject()
    val queryBool = new JSONObject()

    val queryMatch = new JSONObject()
    queryMatch.put(codetype, code)
    val queryMast = new JSONObject()
    queryMast.put("match", queryMatch)
    queryBool.put("must", queryMast)

    val queryTime = new JSONObject()
    queryTime.put("gte", starttime)
    queryTime.put("lte", endtime)
    val querycaptime = new JSONObject()
    querycaptime.put("captime", queryTime)
    val queryRange = new JSONObject()
    queryRange.put("range", querycaptime)
    val queryMasts = new JSONArray()
    queryMasts.add(queryRange)
    val queryBools = new JSONObject()
    queryBools.put("must", queryMasts)
    val queryFilter = new JSONObject()
    queryFilter.put("bool", queryBools)
    queryBool.put("filter", queryFilter)
    queryPara.put("bool", queryBool)
    val esRdd = EsSpark.esRDD(spark.sparkContext, index, queryPara.toJSONString)

    esRdd.map(x => {
      (x._2.get("device_code").get + "", 1)
    })
      .reduceByKey(_ + _) //device,count
      .map(x => {
      x.swap
    })
      .groupByKey() //count,devices
      .collect()
      .sortBy(_._1)
      .reverse
      .splitAt(3) //点位count出现次数拍前三的
      ._1
      .sortBy(tump => {
        //按出现点位数相同的点位数据集个数进行排序
        tump._2.size
      })
      .reverse
      .head //得到出现点位数相同的点位数据集个数最多的
      ._2
      .toArray
  }

}
case class FollowingSiteConditionEntity(idType: String, idNumber: String, beginTime: Long, endTime: Long, slice: Int, siteNumbers: String, hasliferow: String, fuzzyway: String, activityTimes: List[ActivityTime])

case class FollowingConditionEntity(idType: String, idNumber: String, beginTime: Long, endTime: Long, slice: Int, deviceNumbers: String, hasliferow: String, fuzzyway: String, activityTimes: List[ActivityTime])

case class WifiEntityWithLongInsteadOfTimestamp(id: String,
                                                showedIdNumber: String,
                                                idNumber: String,
                                                idNumberType: String,
                                                behaviorStartTime: Long,
                                                behaviorEndTime: Long,
                                                deviceNumber: String,
                                                address: String,
                                                location:String,
                                                devicename: String,
                                                captime: String,
                                                phone:String,
                                                latitude: java.lang.Double,
                                                longitude: java.lang.Double,
                                                picpath:String
                                               )

case class FollowingResult(jobId: String, idNumber: String, idNumberType: String, count: Long, conditions: String)

case class FollowResult(
                         resultid: String,
                         codetype: String,
                         //                         conditions: String,
                         count: Integer,
                         createtime: String,
                         jobid: Integer,
                         tagcode: String,
                         address: String,
                         location:String,
                         phone: String
                       )

case class FollowResultWithDetail(
                                   resultid: String,
                                   codetype: String,
                                   //                         conditions:String,
                                   count: Integer,
                                   createtime: String,
                                   jobid: Integer,
                                   tagcode: String,
                                   address: String,
                                   location:String,
                                   phone: String,
                                   detail: util.ArrayList[FollowDetail] = new util.ArrayList[FollowDetail]()
                                 )

case class FollowDetail(
                         //                       id:Integer,
                         captime: String,
                         codetype: String,
                         devicecode: Integer,
                         devicename: String,
                         address: String,
                         phone: String,
                         var resultid: String,
                         tagcode: String,
                         latitude: java.lang.Double,
                         //                         longitude: java.lang.Double
                         longtitude: java.lang.Double,
                         picpath: String
                       ) {
  def setResultId(string: String) = {
    resultid = string
  }
}
